package net.sudot.flow;

import groovy.lang.Closure;
import groovy.lang.Script;
import lombok.extern.slf4j.Slf4j;
import net.sudot.flow.dsl.model.Context;
import net.sudot.flow.dsl.model.Flow;
import net.sudot.flow.dsl.model.Send;
import net.sudot.flow.dto.ResponseVo;
import net.sudot.flow.utils.GroovyExecuteUtils;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.core.io.ClassPathResource;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.util.Assert;
import org.springframework.web.client.RestTemplate;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.InputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 流程执行器
 *
 * @author tangjialin on 2020-03-16.
 */
@Slf4j
public class FlowActuator implements Callable<Object> {
    public static final String MAIN_SEND_KEY = "main";
    public static final String GROOVY_UTILS_GROOVY = "flow/GroovyUtils.groovy";
    private static final Random RANDOM = new Random();
    private static final ThreadPoolExecutor EXECUTOR;
    private static final String SCRIPT_TEXT;

    static {
        // 构造线程池
        int corePoolSize = 10;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(1000);
        ThreadFactory threadFactory = new DefaultThreadFactory(FlowActuator.class.getSimpleName());
        EXECUTOR = new ThreadPoolExecutor(corePoolSize, corePoolSize, 10L, TimeUnit.SECONDS, workQueue, threadFactory);
        try (InputStream inputStream = new ClassPathResource(GROOVY_UTILS_GROOVY).getInputStream()) {
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            byte[] bytes = new byte[4096];
            for (int len = -1; (len = inputStream.read(bytes)) != -1; ) {
                outputStream.write(bytes, 0, len);
            }
            SCRIPT_TEXT = outputStream.toString();
        } catch (Exception e) {
            throw new RuntimeException(GROOVY_UTILS_GROOVY + "文件读取失败", e);
        }
    }

    private RestTemplate restTemplate = new RestTemplate();
    /** 输入参数 */
    private Map<String, Object> inputData;
    /** 输入参数 */
    private String script;
    private DiscoveryClient discoveryClient;

    public FlowActuator(DiscoveryClient discoveryClient, Map<String, Object> inputData, String script) {
        this.discoveryClient = discoveryClient;
        this.inputData = inputData;
        this.script = script;
    }

    @Override
    public Object call() {
        Map<String, Object> variables = new HashMap<>(0);
        Context context = new Context();
        context.setInputData(inputData);
        context.setVariables(variables);

        // 实例化对象
        Flow flow = new Flow();
        // 加载并实例化 DSL 模型
        String filePath = ClassLoader.getSystemResource(script).getPath();
        Script groovyObject = (Script) GroovyExecuteUtils.loadAndCreateGroovyObject(new File(filePath));
        // 将 DSL 模型关联至 Flow 对象
        try {
            Map<Object, Object> bindingVariables = new HashMap<>(4);
            bindingVariables.put("scriptObject", groovyObject);
            bindingVariables.put("delegateObject", flow);
            GroovyExecuteUtils.execute(SCRIPT_TEXT, bindingVariables);
        } catch (Exception e) {
            throw new RuntimeException(GROOVY_UTILS_GROOVY + "执行错误", e);
        }
        groovyObject.run();

        // 此时的 Flow 对象可以使用了
        log.info("Flow 已装配完毕");

        Map<String, Send> sends = flow.getSends();
        Send mainSend = sends.get(FlowActuator.MAIN_SEND_KEY);
        Assert.notNull(mainSend, "缺少主[" + FlowActuator.MAIN_SEND_KEY + "] send");

        // 执行 pre 函数
        context.setData(call(flow.getPre(), context, inputData));
        // 验证请求参数合法性
        try {
            call(flow.getCheck(), context, null);
        } catch (AssertionError e) {
            return ResponseVo.fail(HttpStatus.BAD_REQUEST.toString(), e.getMessage());
        }
        // 开始执行 send 函数，从名为 main 的 send 函数开始
        execute(MAIN_SEND_KEY, sends, context);

        Object result = null;
        Exception exception = null;
        try {
            // 执行 result 函数
            result = call(flow.getResult(), context, null);
        } catch (Exception e) {
            exception = e;
            log.error("Result函数执行失败", e);
        }
        // 执行 after 函数
        call(flow.getAfter(), context, null, result, exception);
        if (result instanceof ResponseVo) {
            return result;
        } else if (exception == null) {
            return ResponseVo.success(result);
        } else {
            return ResponseVo.fail(HttpStatus.INTERNAL_SERVER_ERROR.toString(), result, exception.getMessage());
        }
    }

    public <T> T call(Closure<T> closure, Context context, T defaultValue, Object... args) {
        if (closure != null) {
            closure.setDelegate(context);
            return closure.call(args);
        }
        return defaultValue;
    }

    protected void execute(String key, Map<String, Send> sends, Context context) {
        Send send = sends.get(key);
        Assert.notNull(send, "Send [" + key + "] 不存在");
        Assert.notNull(send.getHost(), "Send [" + key + "] host 不能为空");
        Assert.hasText(send.getHost(), "Send [" + key + "] host 不能为空");
        Assert.notNull(send.getRequest(), "Send [" + key + "] request 不能为空");
        if (send.isAsync()) {
            EXECUTOR.submit(() -> {
                String nextSend = execute(key, send, context);
                if (nextSend != null) {
                    execute(nextSend, sends, context);
                }
            });
        } else {
            String nextSend = execute(key, send, context);
            if (nextSend != null) {
                execute(nextSend, sends, context);
            }
        }
    }

    protected String execute(String key, Send send, Context context) {
        String url = send.getHost();
        if (!url.contains(":")) {
            try {
                List<ServiceInstance> instances = discoveryClient.getInstances(url);
                if (instances.isEmpty()) {
                    log.error("Send [{}] 执行失败,没有可用的实例:{}", key, url);
                    return null;
                }
                if (instances.size() == 1) {
                    url = instances.get(0).getUri().toString();
                } else {
                    url = instances.get(Math.abs(RANDOM.nextInt()) % instances.size()).getUri().toString();
                }
            } catch (Exception e) {
                if (send.isAsync()) {
                    log.error("Send [{}] 执行失败,服务查询失败:{}", key, url, e);
                    return null;
                }
                Closure<String> fail = send.getFail();
                if (fail != null) {
                    return call(fail, context, null, e);
                }
            }
        }
        String path = send.getPath();
        if (path != null && !path.isEmpty() && !path.startsWith("/")) {
            if (!url.endsWith("/")) {
                url += "/";
            }
        }
        url += path;
        Exception exception = null;
        HttpEntity<?> httpEntity = new HttpEntity<>(call(send.getRequest(), context, null));
        int retry = send.getRetry();
        while (retry >= 0) {
            retry--;
            try {
                ResponseEntity<Object> exchange = restTemplate.exchange(url, send.getMethod(), httpEntity, Object.class);
                Object responseData = call(send.getResponse(), context, exchange.getBody(), exchange);
                // 保存请求结果
                context.getResponseData().put(key, responseData);
                // 判断下一步执行
                return call(send.getNext(), context, null, exchange);
            } catch (Exception e) {
                if (exception == null) {
                    exception = e;
                }
            }
        }
        if (exception != null) {
            if (send.isAsync()) {
                log.error("{} 执行失败", key, exception);
            }
            Closure<String> fail = send.getFail();
            if (fail != null) {
                return call(fail, context, null, exception);
            }
            throw new RuntimeException(exception);
        }
        return null;
    }
}
